package io.kwdb.examples;

import io.kwdb.sdk.*;
import io.kwdb.sdk.internal.Metrics;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.*;

/**
 * Example 03: Batch Processing and Backpressure
 *
 * <p>Demonstrates: - Automatic batching with maxBatchSize and lingerMs - BLOCK backpressure
 * strategy (wait when queue is full) - DROP_OLDEST backpressure strategy (drop old items when full)
 * - Metrics collection and reporting - Multi-threaded write performance
 */
public class Example03_BatcherAndBackpressure {

  private static final Logger logger =
      Logger.getLogger(Example03_BatcherAndBackpressure.class.getName());

  public static void main(String[] args) throws Exception {
    System.out.println("=====================================");
    System.out.println(" KWDB Java SDK M3 - Batcher Example");
    System.out.println("=====================================\n");

    // Configure logging to show batcher activity
    Logger.getLogger("io.kwdb").setLevel(Level.INFO);

    // Test 1: Batching with automatic triggers
    testAutoBatching();

    // Test 2: Backpressure with BLOCK strategy
    testBlockBackpressure();

    // Test 3: Backpressure with DROP_OLDEST strategy
    testDropOldestBackpressure();

    // Test 4: Performance comparison
    testPerformanceComparison();

    System.out.println("\n=====================================");
    System.out.println(" All Batcher Tests Completed");
    System.out.println("=====================================");
  }

  /** Test automatic batching with size and time triggers. */
  private static void testAutoBatching() throws Exception {
    System.out.println("\n--- Test 1: Automatic Batching ---");

    Config config =
        Config.builder()
            .host("127.0.0.1")
            .httpPort(9091)
            .transport(Config.Transport.ILP)
            .maxBatchSize(10) // Batch up to 10 points
            .lingerMs(100) // Wait up to 100ms
            .maxQueueSize(100) // Queue up to 100 points
            .enableRetry(true)
            .build();

    try (KwdbClient client = KwdbClient.create(config)) {
      System.out.println("Config: maxBatchSize=10, lingerMs=100ms");

      // Write 25 points rapidly
      for (int i = 0; i < 25; i++) {
        Point point =
            Point.builder("batch_test")
                .tag("host", "batch-host-" + i)
                .tag("type", "auto")
                .field("value", i * 1.5)
                .timestampNow()
                .build();

        Status status = client.write(point);
        if (!status.isOk()) {
          System.err.println("Write failed: " + status.getMessage());
        }

        // Add small delay every 10 points
        if (i > 0 && i % 10 == 0) {
          Thread.sleep(50);
          System.out.println("Wrote " + i + " points...");
        }
      }

      // Flush remaining points
      System.out.println("Flushing remaining points...");
      client.flush();

      // Get metrics
      Metrics.MetricsSnapshot metrics = Metrics.getInstance().getSnapshot();
      System.out.println("✓ Writes successful: " + metrics.writesOk);
      System.out.println("✓ Batch latency (p50): " + metrics.batchLatency.p50 + "ms");
      System.out.println("✓ Batch latency (p95): " + metrics.batchLatency.p95 + "ms");
    }
  }

  /** Test BLOCK backpressure strategy. */
  private static void testBlockBackpressure() throws Exception {
    System.out.println("\n--- Test 2: BLOCK Backpressure ---");

    Config config =
        Config.builder()
            .host("127.0.0.1")
            .httpPort(9091)
            .transport(Config.Transport.ILP)
            .maxBatchSize(5)
            .lingerMs(50)
            .maxQueueSize(10) // Small queue to trigger backpressure
            .backpressure(Config.Backpressure.BLOCK)
            .timeoutMillis(1000) // Wait up to 1s when blocked
            .build();

    try (KwdbClient client = KwdbClient.create(config)) {
      System.out.println("Config: maxQueueSize=10, backpressure=BLOCK");

      ExecutorService executor = Executors.newFixedThreadPool(5);
      CountDownLatch latch = new CountDownLatch(50);
      AtomicInteger blocked = new AtomicInteger(0);
      AtomicInteger succeeded = new AtomicInteger(0);

      // Submit 50 writes from 5 threads
      for (int i = 0; i < 50; i++) {
        final int id = i;
        executor.submit(
            () -> {
              try {
                Point point =
                    Point.builder("backpressure_test")
                        .tag("thread", Thread.currentThread().getName())
                        .tag("strategy", "block")
                        .field("id", id)
                        .timestampNow()
                        .build();

                Status status = client.write(point);
                if (status.isOk()) {
                  succeeded.incrementAndGet();
                } else if (status.getCode() == Status.Code.RESOURCE_EXHAUSTED) {
                  blocked.incrementAndGet();
                  System.out.println("Write " + id + " blocked/rejected");
                }
              } finally {
                latch.countDown();
              }
            });
      }

      // Wait for all writes to complete
      latch.await(10, TimeUnit.SECONDS);
      executor.shutdown();

      // Flush remaining
      client.flush();

      System.out.println("✓ Succeeded: " + succeeded.get());
      System.out.println("✓ Blocked/Rejected: " + blocked.get());
      System.out.println("✓ Total: " + (succeeded.get() + blocked.get()));
    }
  }

  /** Test DROP_OLDEST backpressure strategy. */
  private static void testDropOldestBackpressure() throws Exception {
    System.out.println("\n--- Test 3: DROP_OLDEST Backpressure ---");

    Config config =
        Config.builder()
            .host("127.0.0.1")
            .httpPort(9091)
            .transport(Config.Transport.ILP)
            .maxBatchSize(5)
            .lingerMs(50)
            .maxQueueSize(10) // Small queue to trigger drops
            .backpressure(Config.Backpressure.DROP_OLDEST)
            .build();

    try (KwdbClient client = KwdbClient.create(config)) {
      System.out.println("Config: maxQueueSize=10, backpressure=DROP_OLDEST");

      // Reset metrics
      Metrics.getInstance().reset();

      // Write many points rapidly to trigger drops
      int totalWrites = 100;
      for (int i = 0; i < totalWrites; i++) {
        Point point =
            Point.builder("drop_test")
                .tag("host", "drop-host-" + i)
                .tag("strategy", "drop_oldest")
                .field("value", i)
                .timestampNow()
                .build();

        client.write(point);
        // No delay - write as fast as possible
      }

      // Flush
      client.flush();

      // Get metrics
      Metrics.MetricsSnapshot metrics = Metrics.getInstance().getSnapshot();
      System.out.println("✓ Total writes attempted: " + totalWrites);
      System.out.println("✓ Writes successful: " + metrics.writesOk);
      System.out.println("✓ Items dropped: " + metrics.droppedItems);
      System.out.println("✓ Drop rate: " + (metrics.droppedItems * 100.0 / totalWrites) + "%");
    }
  }

  /** Compare performance with and without batching. */
  private static void testPerformanceComparison() throws Exception {
    System.out.println("\n--- Test 4: Performance Comparison ---");

    int numPoints = 100;

    // Test without batching (direct write)
    System.out.println("\nDirect Write Mode (no batching):");
    Config directConfig =
        Config.builder()
            .host("127.0.0.1")
            .httpPort(9091)
            .transport(Config.Transport.ILP)
            .maxBatchSize(0) // Disabled
            .enableRetry(false)
            .build();

    long directStart = System.currentTimeMillis();
    try (KwdbClient client = KwdbClient.create(directConfig)) {
      for (int i = 0; i < numPoints; i++) {
        Point point =
            Point.builder("perf_test")
                .tag("mode", "direct")
                .field("value", i)
                .timestampNow()
                .build();
        client.write(point);
      }
    }
    long directElapsed = System.currentTimeMillis() - directStart;
    System.out.println("✓ Time for " + numPoints + " points: " + directElapsed + "ms");
    System.out.println("✓ Throughput: " + (numPoints * 1000.0 / directElapsed) + " points/sec");

    // Test with batching
    System.out.println("\nBatch Mode (maxBatchSize=50):");
    Config batchConfig =
        Config.builder()
            .host("127.0.0.1")
            .httpPort(9091)
            .transport(Config.Transport.ILP)
            .maxBatchSize(50)
            .lingerMs(10)
            .maxQueueSize(500)
            .enableRetry(false)
            .build();

    long batchStart = System.currentTimeMillis();
    try (KwdbClient client = KwdbClient.create(batchConfig)) {
      for (int i = 0; i < numPoints; i++) {
        Point point =
            Point.builder("perf_test")
                .tag("mode", "batch")
                .field("value", i)
                .timestampNow()
                .build();
        client.write(point);
      }
      client.flush();
    }
    long batchElapsed = System.currentTimeMillis() - batchStart;
    System.out.println("✓ Time for " + numPoints + " points: " + batchElapsed + "ms");
    System.out.println("✓ Throughput: " + (numPoints * 1000.0 / batchElapsed) + " points/sec");

    // Compare
    double speedup = (double) directElapsed / batchElapsed;
    System.out.println("\n✓ Speedup with batching: " + String.format("%.2fx", speedup));
  }
}
